/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.TreeSet;

import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;
import cgl.narada.service.reliable.events.RdNakInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdRetransmissionServiceEvent;

/**
   This class is used to ensure that Naks are issued such that the number of
   retransmissions is very controlled. At any given time for a given template
   there should be at the most 2 Naks that are operational. We refer to this 
   as previousNak and currentNak.
   
   @author Shrideep Pallickara
   $Date$
   $Revision$
*/

public class NakProcessor implements RdDebugFlags {

  private int entityId;

  /** templateId -> Nak*/
  private Hashtable previousNaks;

  /** templateId -> Nak*/
  private Hashtable currentNaks;

  /** templateId -> TreeSet  missedSequences*/
  private Hashtable pendingSequences;

  private int MAX_SEQUENCES_IN_NAK = 5;
  private EventProducer producer;
  private String rdsString = "ReliableDeliveryService/";
  
  private String moduleName = "NakProcessor: ";

  public NakProcessor(int entityId, EventProducer producer) {
    this.entityId = entityId;
    this.producer = producer;
    currentNaks = new Hashtable();
    previousNaks = new Hashtable();
    pendingSequences = new Hashtable();
  }
  
  
  /** We have to check if the reported missedSequence is contained in 
      either a previously issued Nak or the currentNak that was the last
      issued Nak. */  
  public void processMissedSequences(int templateId, long[] missedSequences) {
    Object templateKey = new Integer(templateId);
    TreeSet missedSet = new TreeSet();
    
    for (int i=0; i<missedSequences.length; i++) {
      missedSet.add( new Long(missedSequences[i]) );
    }
        
    processMissedSet(templateKey, missedSet);
  }  
  
  
  public void processMissedSequence(int templateId, long missedSequence) {
    Object templateKey = new Integer(templateId);
    TreeSet missedSet = new TreeSet();
    missedSet.add( new Long(missedSequence) );
    
    processMissedSet(templateKey, missedSet);
  }

  private void processMissedSet(Object templateKey, TreeSet missedSet) {
    if (!previousNaks.containsKey(templateKey) && 
	!currentNaks.containsKey(templateKey) ) {
      /** This implies that this would be the first Nak that we need to 
	  issue */
      long[] nakSequences;
      int numOfSeqInNak = 0;
      if (missedSet.size() <= MAX_SEQUENCES_IN_NAK) {
	numOfSeqInNak = missedSet.size();
      } else {
	numOfSeqInNak = MAX_SEQUENCES_IN_NAK;
      }
      
      nakSequences = new long[numOfSeqInNak];
      
      for (int i=0; i< numOfSeqInNak; i++) {
	nakSequences[i] = ((Long) missedSet.first()).longValue();
	missedSet.remove(missedSet.first());
      }
      createAndIssueNak(((Integer)templateKey).intValue(), nakSequences);
      
    } /** both previous and current donot contain Nak */
    
    
    
    /** Search to see if the nakSequence is contained in the missedSet.
	If it is we need to remove it from the missedSet */
    if (previousNaks.containsKey(templateKey)) {
      RdNakInvoiceEntityEvent previousNak = 
	(RdNakInvoiceEntityEvent) previousNaks.get(templateKey);
      if (previousNak.containsSingleNak()) {
	long nakSequence = previousNak.getNakSequenceNumber();
	processNakSequenceAndMissedSet(nakSequence, missedSet);
      } else {
	long[] nakSequences = previousNak.getNakSequenceNumbers();
	
	for (int i=0; i < nakSequences.length; i++) {
	  processNakSequenceAndMissedSet(nakSequences[i], missedSet);
	}
      } /* end if containsSingleNak() */
      
    } /* end if contained in previous Nak */

    
    /** Search to see if the nakSequence is contained in the missedSet.
	If it is we need to remove it from the missedSet */
    if (currentNaks.containsKey(templateKey)) {
      RdNakInvoiceEntityEvent currentNak = 
	(RdNakInvoiceEntityEvent) currentNaks.get(templateKey);
      if (currentNak.containsSingleNak()) {
	long nakSequence = currentNak.getNakSequenceNumber();
	processNakSequenceAndMissedSet(nakSequence, missedSet);
      } else {
	long[] nakSequences = currentNak.getNakSequenceNumbers();
	for (int i=0; i < nakSequences.length; i++) {
	  processNakSequenceAndMissedSet(nakSequences[i], missedSet);
	}
      } /* end if containsSingleNak() */
      
    } /* end if contained in current Nak */

    
    if (missedSet.size() == 0) {
      return;
    }

    /** Now that we are here, the missedVec contains only those sequences 
	which are not present in either the previousNaks or the currentNaks */
    if (!pendingSequences.containsKey(templateKey)) {
      pendingSequences.put(templateKey, new TreeSet());
    }
    
    TreeSet pendingSet = (TreeSet) pendingSequences.get(templateKey);
    pendingSet.addAll(missedSet);
    if (NakProcessor_Debug) {
      System.out.println(moduleName + "Data in pending Set = " + pendingSet);
    }
  }

   /** Checks to see if a given nakSequence already exists in a missedSet
      and if it is, this nakSequence is removed from the missedVec. */
  private void processNakSequenceAndMissedSet(long nakSequence, 
					      TreeSet missedSet) {
    Long seqObj = new Long(nakSequence);
    if (missedSet.contains(seqObj))
      missedSet.remove(seqObj);
  }


  


  public void processSyncAdvance(int templateId, long syncAdvance) {
    Object templateKey = new Integer(templateId);
    
    if (!pendingSequences.containsKey(templateKey) ) {
      if (NakProcessor_Debug) {
	//System.out.println(moduleName+"Unware of templateId=" + templateId + 
	//  ". Cannot process syncAdvance to->" + syncAdvance);
      }
      pendingSequences.put(templateKey, new TreeSet());
      return;
    }
    
    TreeSet pendingSet = (TreeSet) pendingSequences.get(templateKey);
    
    /** Removes sequences from the pending set that are lesser/equal to 
	the syncAdvance */
    for (int i=0; i < pendingSet.size(); i++) {
      long compare = ((Long) pendingSet.first()).longValue();
      
      if (compare > syncAdvance) {
	break;
      } else {
	pendingSet.remove(pendingSet.first());
      }
      
    }/* end for */     
  }
  



  


  public void 
  processRetransmittedEvent(RdRetransmissionServiceEvent retransmitEvent) { 
    RdArchivalServiceNotification archivalEvent = 
      retransmitEvent.getArchivalNotification();
    
    long retransmitId = retransmitEvent.getRetransmissionId();
    long sequenceNumber = archivalEvent.getSequenceNumber();
    int templateId = archivalEvent.getTemplateId();

    Object templateKey = new Integer(templateId);
    
    /** If the response is for a previousNak, check to see if the transmissions
	have been satisfied for the previousNak, and proceed to remove the
	entry for the previousNak */
    if (previousNaks.containsKey(templateKey) ) {
      RdNakInvoiceEntityEvent previousNak = 
	(RdNakInvoiceEntityEvent) previousNaks.get(templateKey);
      long previousId = previousNak.getRetransmissionId();
      
      if (previousId == retransmitId) {
	previousNak.removeNakSequence(sequenceNumber);

	if (!previousNak.containsNakInfo()) {
	  previousNaks.remove(templateKey);
	}
	
	return;
      }/*end if (previousId == retransmitId) */
      
    }

    
    /** If the response is for a currentNak and there is no previousNak,
	it is time to issue a new Nak with a new set of nakSequences */
    if (currentNaks.containsKey(templateKey) ) {
      RdNakInvoiceEntityEvent currentNak = 
	(RdNakInvoiceEntityEvent) currentNaks.get(templateKey);
      long currentId = currentNak.getRetransmissionId();
      
      
      if (currentId == retransmitId) {
	currentNak.removeNakSequence(sequenceNumber);
	if (!currentNak.containsNakInfo()) {
	  currentNaks.remove(templateKey);
	}
	
	/** If there is no previous Nak it is time to issue a new Nak with a 
	    new set of nakSequences */
	if (!previousNaks.containsKey(templateKey)) {
	  proceedToCreateAndIssueNak(templateKey);
	  return;
	}
	
	/** If the response is not for the previousNak and if some of the
	    sequences in the previousNak are still pending, it implies that
	    either the Nak or some of the retransmissions were lost in 
	    transit.
	    We create a new set of nakSequences which includes sequences 
	    pending from the previousNak and proceed to issue a new Nak. */
	RdNakInvoiceEntityEvent previousNak = 
	  (RdNakInvoiceEntityEvent) previousNaks.get(templateKey);
	TreeSet pendingSet = (TreeSet) pendingSequences.get(templateKey);
	if (previousNak.containsNakInfo()) {
	  if (previousNak.containsSingleNak()) {
	    pendingSet.add( new Long(previousNak.getNakSequenceNumber()) );
	  } else {
	    long[] nakSeqs = previousNak.getNakSequenceNumbers();
	    for (int i=0; i< nakSeqs.length; i++) {
	      pendingSet.add(new Long(nakSeqs[i]) );
	    }/*end for */
	  } /*end if containsSingleNak*/
	  proceedToCreateAndIssueNak(templateKey);
	}/*end if containsNakInfo*/
	return;
      } /*end if (currentId == retransmitId) */
            
    }/*end if (currentNaks.containsKey(templateKey)*/
    
    System.out.println(moduleName + "No idea how to process received " +
		       "retransmission event " + retransmitEvent);
  }
  
  private void createAndIssueNak(int templateId, long[] nakSequences) {
    RdNakInvoiceEntityEvent nakInvoice;
    Object templateKey = new Integer(templateId);

    if (nakSequences.length == 1) {
      nakInvoice = 
	new RdNakInvoiceEntityEvent(templateId, entityId, nakSequences[0]);
    } else {
      nakInvoice = 
	new RdNakInvoiceEntityEvent(templateId, entityId, nakSequences);
    } /* end if (nakSequences.length == 1) */
    
    if (currentNaks.containsKey(templateKey)) {
      previousNaks.remove(templateKey);
      previousNaks.put(templateKey, currentNaks.remove(templateKey));
    }
    
    currentNaks.put(templateKey, nakInvoice);
    issueRetransmissionRequest(nakInvoice);
  }
  
  private void proceedToCreateAndIssueNak(Object templateKey) {
    TreeSet pendingSet = (TreeSet) pendingSequences.get(templateKey);

    if (pendingSet == null) 
      return;
    
    if (pendingSet.size() == 0)
      return;
    
    int numOfSeqInNak = 0;
    
    if (pendingSet.size() > MAX_SEQUENCES_IN_NAK) {
      numOfSeqInNak = MAX_SEQUENCES_IN_NAK;
    } else {
      numOfSeqInNak = pendingSet.size();
    }
    
    if (numOfSeqInNak == 0) {
      return;
    }

    long[] nakSequences = new long[numOfSeqInNak];
    for (int i=0; i< numOfSeqInNak; i++) {
      nakSequences[i] = ((Long) pendingSet.first()).longValue();
      pendingSet.remove(pendingSet.first());
    }
    createAndIssueNak(((Integer)templateKey).intValue(), nakSequences);
  }

  /** Issues a retransmission request */
  private void issueRetransmissionRequest(RdNakInvoiceEntityEvent nakInvoice) {
    try {
      if (NakProcessor_Debug) {
	System.out.println("\n" + moduleName + 
			   "Issuing Retransmission request " + nakInvoice 
			   + "\n\n");
      }
      int templateId = nakInvoice.getTemplateId();
      NBEvent nbEvent = 
	producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
			       rdsString + templateId + "/Invoice", 
			       nakInvoice.getBytes());
      
      NBEventGenerator.setEventType(nbEvent, 
				    NBEventTypes.RELIABLE_DELIVERY_EXCHANGE);
      producer.publishEvent(nbEvent);
    } catch (ServiceException serEx) {
      System.out.println("\n\n\n"+ moduleName + serEx +"\n\n\n");
    }
    
  }

  /** Indicates if there are pending Naks to issue. We need to cycle through 
      the pending sets associated with every templateId and then check if there
      are any pending sequences. */
  public boolean hasPendingNaksToIssue() {
    
    if (pendingSequences.size() == 0) {
      return false;
    }

    Enumeration e = pendingSequences.elements();

    while (e.hasMoreElements()) {
      TreeSet _pendingTemplateSet = (TreeSet) e.nextElement();
      if (_pendingTemplateSet.size() > 0) {
	return true;
      }
    }/* end while */
    
    return false;
  }


  public void proceedToIssuePendingNaks() {
    if (pendingSequences.size() == 0) {
      return;
    }
    
    Enumeration e = pendingSequences.keys();

    while (e.hasMoreElements()) {
      Object _templateKey = e.nextElement();
      proceedToCreateAndIssueNak(_templateKey);
    }/*end while */
  }

}




